package com.zhb.cloud.distribute.zk.lock;

import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.locks.InterProcessLock;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import com.zhb.cloud.distribute.lock.DistributedLock;
import com.zhb.cloud.distribute.lock.LockException;

/**
 * @ClassName: ZooKeeperDistributedLock
 * @Description:zookeeper 分布式锁实现
 * @author: zhb zhb1316783262@163.com
 * @date: 2017年11月20日 下午3:39:46
 *
 */
public class ZooKeeperDistributedLock implements DistributedLock {

	private static final Logger LOG = LoggerFactory.getLogger(ZooKeeperDistributedLock.class);
	@Autowired
	private CuratorFramework client;
	/*
	 * 锁路径--(粒度)
	 */
	private String lockPath;
	/**
	 * 分布式zookeeper锁
	 */
	private InterProcessLock interProcessLock;
	/**
	 * zookeeper 客户端缓存 ConcurrentHashMap来做
	 */
	private static final ConcurrentHashMap<String, CuratorFramework> sZKClientCache = new ConcurrentHashMap<String, CuratorFramework>();

	static {
		// shut down all zookeeper clients when exits
		Runtime.getRuntime().addShutdownHook(new Thread() {
			@Override
			public void run() {
				// <Configuration shutdownHook="disable"> in log4j2.xml to
				// support logger here
				closeZooKeeperClients();
			}
		});
	}

	/**
	 * Close zookeeper client
	 */
	private static synchronized void closeZooKeeperClients() {
		for (Entry<String, CuratorFramework> entry : sZKClientCache.entrySet()) {
			entry.getValue().close();
			sZKClientCache.remove(entry.getKey());
		}
		LOG.info("zookeeper client destroy succeed.");
	}

	/*
	 * 构造函数初始化分布式锁
	 */
	public ZooKeeperDistributedLock(String lockPath) {
		if (client == null) {
			throw new LockException("no zookeeper client found");
		}
		this.lockPath = lockPath;
		this.interProcessLock = new InterProcessMutex(client, lockPath);
	}

	/**
	 * 加锁
	 */
	@Override
	public void lock() throws LockException {
		try {
			this.interProcessLock.acquire();
			LOG.info("{} acquire lock: {}", Thread.currentThread(), lockPath);
		} catch (Exception e) {
			throw new LockException("failed to acquire the lock", e);
		}
	}

	/**
	 * 加锁
	 */
	@Override
	public void lock(long time, TimeUnit unit) throws LockException {
		try {
			this.interProcessLock.acquire();
			LOG.info("{} acquire lock: {}", Thread.currentThread(), lockPath);
		} catch (Exception e) {
			throw new LockException("failed to acquire the lock", e);
		}
	}

	/**
	 * 释放锁
	 */
	@Override
	public void unlock() throws LockException {
		try {
			this.interProcessLock.release();
			LOG.info("{} release lock: {}", Thread.currentThread(), lockPath);
		} catch (Exception e) {
			throw new LockException("failed to release the lock", e);
		}
	}

	/**
	 * 加锁
	 */
	@Override
	public boolean tryLock() throws LockException {
		return tryLock(1, TimeUnit.MILLISECONDS);
	}

	/**
	 * 加锁
	 */
	@Override
	public boolean tryLock(long time, TimeUnit unit) throws LockException {
		boolean result = false;
		try {
			result = this.interProcessLock.acquire(time, unit);
		} catch (Exception e) {
			throw new LockException("failed to try acquire the lock", e);
		}
		LOG.info("{} try lock: {} result: {}", Thread.currentThread(), this.lockPath, result);
		return result;
	}

	/**
	 * 加锁
	 */
	@Override
	public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) {
		return tryLock(waitTime, unit);
	}

	/**
	 * 是否锁定
	 */
	@Override
	public boolean isLocked() throws LockException {
		return this.interProcessLock.isAcquiredInThisProcess();
	}
}